package com.source.pekko;

import org.apache.pekko.actor.AbstractActor;
import org.apache.pekko.actor.ActorRef;
import org.apache.pekko.japi.pf.ReceiveBuilder;

/**
 * 继承AbstractActor定义自己的actor
 * Actor可以发送和接收消息
 */
public class PekkoRpcActor extends AbstractActor {
    /**
     * 实现接收消息
     * @return
     */
    @Override
    public Receive createReceive() {
        return ReceiveBuilder.create()
                /**接收到PekkoData消息交给handleMessage处理
                 * flink PekkoRpcActor 155行也是这样处理的
                 */
                .match(PekkoData.class, this::handleMessage)
                .build();
    }

    /**
     * 处理具体消息
     * @param message
     */
    private void handleMessage(final PekkoData message) {
        /** 获取发送者，发送者对应的就是actorRef */
        ActorRef sender = getSender();
        ActorRef self = getSelf();
        /** 打印 */
        System.out.println(sender + ":发送者=>" + message.getInfo());
        /** 回复消息 向发送者sender 回复word 的消息 回复者是当前actorRef*/
        /** 4、Actor 可以改变他自身的状态，可以接收消息，也可以发送消息，还可以生成新的 Actor  */
        sender.tell(new PekkoData("PekkoRpcActor-word"),self);

    }
}
